/** @file ar_rpc.c
  *
  * @brief This file contains the artosyn rpc function calls. Waiting
  * for move daemon here.
  *
  *  Copyright (C) 2024, Artosyn Technologies. ALL RIGHTS RESERVED.
  *
  *  This Information is proprietary to Artosyn Technologies and MAY NOT
  *  be copied by any method or incorporated into another program without
  *  the express written consent of Artosyn. This Information or any portion
  *  thereof remains the property of Artosyn. The Information contained herein
  *  is believed to be accurate and Artosyn assumes no responsibility or
  *  liability for its use in any way and conveys no license or title under
  *  any patent or copyright and makes no representation or warranty that this
  *  Information is free from patent or copyright infringement.
  *
  */

/****************************
 * Include
 ****************************/
#include <linux/kernel.h>
#include <linux/netdevice.h>
#include <linux/module.h>
#include <linux/vmalloc.h>

#include "utils/utils.h"
#include "rpc/ar_rpc.h"
#include "rpc/session_socket.h"
#include "net/net_dev.h"
#include "oal_main.h"
#include "utils/ringbuffer.h"

int get_fix_usblen(void)
{
    return fixpackbase;
}

int get_usb_data_offset(void)
{
    return data_offset;
}

static u8 xor_check(unsigned char* buff, int len)
{
    int i;
    u8 xor = 0xff;
    for (i = 0; i < len; i++) {
        xor ^= buff[i];
    }
    return xor;
}

static u8 xor_check_rb(ringbuffer_t *rb, int offset, int len)
{
    int i;
    u8 xor = 0xff;
    u8 *buff = NULL;

    if (!rb) {
        return 0xff;
    }

    if (ringbuffer_len(rb) < len) {
        return 0xff;
    }

    buff = rb->buffer;

    for (i = 0; i < len; i++) {
        xor ^= buff[RB_POS_MOV(rb, rb->read, offset + i)];
    }

    return xor;
}

static int make_datagram_pack2buff(unsigned char* buff, u32 len, rpc_pack_t* pack)
{
    u32 maxlen = fixpackbase + pack->datalen;

    if (len < maxlen) {
        return -1;
    }

    int offset = 0;

    buff[offset++] = 0xab;

    buff[offset++] = pack->datalen & 0xff;
    buff[offset++] = (pack->datalen >> 8) & 0xff;
    buff[offset++] = (pack->datalen >> 16) & 0xff;
    buff[offset++] = (pack->datalen >> 24) & 0xff;

    u8 orchk  = xor_check(buff, offset);
    buff[offset++] = orchk;

    memcpy(buff + offset, pack->datapack, pack->datalen);
    offset += pack->datalen;

    buff[offset++] = 0xbc;

    return maxlen;
}

/**
 * @brief Assemble the rpc pkt. Headrom and tailroom should be reserved!
 *
 * @param   buff   Point to the buffer
 * @param   len    Length of the buffer
 *
 * @return  0 for success, otherwise fail.
 */
int
make_datagram_pack2buff_nocopy(unsigned char* buff, u32 len)
{
    int offset = 0;

    buff[offset++] = 0xab;

    buff[offset++] = len & 0xff;
    buff[offset++] = (len >> 8) & 0xff;
    buff[offset++] = (len >> 16) & 0xff;
    buff[offset++] = (len >> 24) & 0xff;

    u8 orchk  = xor_check(buff, offset);
    buff[offset++] = orchk;

    offset += len;

    buff[offset++] = 0xbc;

    return len + datagram_len;
}

/**
 * @brief 从二进制中找出socket datapram
 *
 * @param buff
 * @param bufflen
 * @param pack
 * @return int
 */
int unpack_socket_datagram_pack(unsigned char* buff, u32 bufflen, u32 *payload_len)
{
    u32 offset;
    if (!buff || bufflen < datagram_len) {
        return -1;
    }

    for (offset = 0; offset <= bufflen - datagram_len; offset++) {
        if (buff[offset + 0] == 0xab) {
            // pack len
            u32 datalen = 0;
            datalen |= buff[offset + 1] << 0;
            datalen |= buff[offset + 2] << 8;
            datalen |= buff[offset + 3] << 16;
            datalen |= buff[offset + 4] << 24;

            u32 maxlen = datagram_len + datalen;

            // chk len
            if (offset + maxlen > bufflen) {
                continue;
            }

            // chk tail
            if (buff[offset + maxlen - 1] != 0xbc) {
                continue;
            }

            // chk len
            u8 cal_xor = xor_check(buff + offset, datagram_offset - 1);

            if (cal_xor != buff[offset + datagram_offset - 1]) {
                continue;
            }

            if (payload_len) {
                *payload_len = datalen;
            }

            return 0;
        }
    }
    return -2;
}

/**
 * @brief 从ringbuffer中找出socket datapram
 *
 * @param rb
 * @param start
 * @param len
 * @return int
 */
static int unpack_socket_datagram_pack_rb(ringbuffer_t *rb, u32 *start, u32 *len)
{
    u32 offset;
    u8 *buff = NULL;

    if (!rb) {
        AR_DBG(DM_RPC, DL_ERR, "%s %d rb is null!!\n",__FUNCTION__,__LINE__);
        return -1;
    }

    if (ringbuffer_len(rb) < datagram_len) {
        AR_DBG(DM_RPC, DL_VRB, "%s %d rb len %d is less than datagram_len %d!!\n",__FUNCTION__,__LINE__, ringbuffer_len(rb), datagram_len);
        return -2;
    }

    buff = rb->buffer;

    for (offset = 0; offset <= ringbuffer_len(rb) - datagram_len; offset++) {
        if (buff[RB_POS_MOV(rb, rb->read, offset + 0)] == 0xab) {
            // pack len
            u32 datalen = 0;
            datalen |= buff[RB_POS_MOV(rb, rb->read, offset + 1)] << 0;
            datalen |= buff[RB_POS_MOV(rb, rb->read, offset + 2)] << 8;
            datalen |= buff[RB_POS_MOV(rb, rb->read, offset + 3)] << 16;
            datalen |= buff[RB_POS_MOV(rb, rb->read, offset + 4)] << 24;

            u32 maxlen = datagram_len + datalen;

            // chk len
            if (offset + maxlen > ringbuffer_len(rb)) {
                AR_DBG(DM_RPC, DL_VRB, "%s %d offset + maxlen %d > ringbuffer_len(rb) %d\n",__FUNCTION__,__LINE__, offset + maxlen, ringbuffer_len(rb));
                continue;
            }

            // chk tail
            if (buff[RB_POS_MOV(rb, rb->read, offset + maxlen - 1)] != 0xbc) {
                AR_DBG(DM_RPC, DL_VRB, "%s %d buff[RB_POS_MOV(rb, offset, maxlen - 1)] 0x%x\n",__FUNCTION__,__LINE__, buff[RB_POS_MOV(rb, rb->read, offset + maxlen - 1)]);
                continue;
            }

            // chk xor
            u8 cal_xor = xor_check_rb(rb, offset, datagram_offset - 1);

            if (cal_xor != buff[RB_POS_MOV(rb, rb->read, offset + datagram_offset - 1)]) {
                AR_DBG(DM_RPC, DL_VRB, "%s %d cal_xor 0x%x != data 0x%x\n",__FUNCTION__,__LINE__,
                        cal_xor,
                        buff[RB_POS_MOV(rb, rb->read, offset + datagram_offset - 1)]);
                continue;
            }

            if (start) {
                *start = offset;
            }

            if (len) {
                *len = datagram_len + datalen;
            }

            return 0;
        }
    }
    return -2;
}

int make_rpc_pack2buff(unsigned char* buff, u32 len, rpc_pack_t* pack)
{
    u32 maxlen = fixpackbase + pack->datalen;

    if (len < maxlen) {
        return -1;
    }

    int offset = 0;

    buff[offset++] = 0xaa;

    buff[offset++] = pack->datalen & 0xff;
    buff[offset++] = (pack->datalen >> 8) & 0xff;
    buff[offset++] = (pack->datalen >> 16) & 0xff;
    buff[offset++] = (pack->datalen >> 24) & 0xff;

    buff[offset++] = pack->reqid >> 24;
    buff[offset++] = pack->reqid >> 16;
    buff[offset++] = pack->reqid >> 8;
    buff[offset++] = pack->reqid >> 0;

    buff[offset++] = pack->msgid >> 24;
    buff[offset++] = pack->msgid >> 16;
    buff[offset++] = pack->msgid >> 8;
    buff[offset++] = pack->msgid >> 0;

    buff[offset++] = pack->sta >> 24;
    buff[offset++] = pack->sta >> 16;
    buff[offset++] = pack->sta >> 8;
    buff[offset++] = pack->sta >> 0;

    u8 orchk  = xor_check(buff, offset);
    buff[offset++] = orchk;

    memcpy(buff + offset, pack->datapack, pack->datalen);
    offset += pack->datalen;

    buff[offset++] = 0xbb;

    return maxlen;
}

int make_rpc_pack2buff_head_only(unsigned char* buff, u32 len, rpc_pack_t* pack)
{
    u32 maxlen = fixpackbase + pack->datalen;

    if (len < maxlen) {
        return -1;
    }

    int offset = 0;

    buff[offset++] = 0xaa;

    buff[offset++] = pack->datalen & 0xff;
    buff[offset++] = (pack->datalen >> 8) & 0xff;
    buff[offset++] = (pack->datalen >> 16) & 0xff;
    buff[offset++] = (pack->datalen >> 24) & 0xff;

    buff[offset++] = pack->reqid >> 24;
    buff[offset++] = pack->reqid >> 16;
    buff[offset++] = pack->reqid >> 8;
    buff[offset++] = pack->reqid >> 0;

    buff[offset++] = pack->msgid >> 24;
    buff[offset++] = pack->msgid >> 16;
    buff[offset++] = pack->msgid >> 8;
    buff[offset++] = pack->msgid >> 0;

    buff[offset++] = pack->sta >> 24;
    buff[offset++] = pack->sta >> 16;
    buff[offset++] = pack->sta >> 8;
    buff[offset++] = pack->sta >> 0;

    u8 orchk  = xor_check(buff, offset);
    buff[offset++] = orchk;

    offset += pack->datalen;

    buff[offset++] = 0xbb;

    return maxlen;
}

/* headrom and tailroom should be reserved!!!!!! */
int make_rpc_pack2buff_no_copy(unsigned char* buff, u32 len, rpc_pack_t* pack)
{
    int offset = 0;

    buff[offset++] = 0xaa;

    buff[offset++] = len & 0xff;
    buff[offset++] = (len >> 8) & 0xff;
    buff[offset++] = (len >> 16) & 0xff;
    buff[offset++] = (len >> 24) & 0xff;

    buff[offset++] = pack->reqid >> 24;
    buff[offset++] = pack->reqid >> 16;
    buff[offset++] = pack->reqid >> 8;
    buff[offset++] = pack->reqid >> 0;

    buff[offset++] = pack->msgid >> 24;
    buff[offset++] = pack->msgid >> 16;
    buff[offset++] = pack->msgid >> 8;
    buff[offset++] = pack->msgid >> 0;

    buff[offset++] = pack->sta >> 24;
    buff[offset++] = pack->sta >> 16;
    buff[offset++] = pack->sta >> 8;
    buff[offset++] = pack->sta >> 0;

    u8 orchk  = xor_check(buff, offset);
    buff[offset++] = orchk;

    offset += len;

    buff[offset++] = 0xbb;

    return len + fixpackbase;
}

/**
 * @brief 从二进制中找出usbpack
 *
 * @param buff
 * @param bufflen
 * @param usedlen 发现到的usbpack最后一个字节的位置
 * @param pack
 * @return int
 */
int unpack_usb_pack(unsigned char* buff, u32 bufflen, int* usedlen, rpc_pack_t* pack)
{
    u32 offset;
    if (!buff || bufflen < fixpackbase || !pack) {
        return -1;
    }

    for (offset = 0; offset <= bufflen - fixpackbase; offset++) {
        if (buff[offset + 0] == 0xaa) {
            // pack len
            u32 datalen = 0;
            datalen |= buff[offset + 1] << 0;
            datalen |= buff[offset + 2] << 8;
            datalen |= buff[offset + 3] << 16;
            datalen |= buff[offset + 4] << 24;

            u32 maxlen = fixpackbase + datalen;

            // chk len
            if (offset + maxlen > bufflen) {
                continue;
            }

            // chk tail
            if (buff[offset + maxlen - 1] != 0xbb) {
                continue;
            }

            // chk len
            u8 cal_xor = xor_check(buff + offset, fixpackbase - 2);

            if (cal_xor != buff[offset + fixpackbase - 2]) {
                continue;
            }

            if (usedlen) {
                *usedlen = offset + maxlen;
            }

            pack->datalen = datalen;
            // head + datalen
            int reqoffset = offset + 5;

            pack->reqid = buff[reqoffset + 0] << 24;
            pack->reqid |= buff[reqoffset + 1] << 16;
            pack->reqid |= buff[reqoffset + 2] << 8;
            pack->reqid |= buff[reqoffset + 3] << 0;

            pack->msgid = buff[reqoffset + 4] << 24;
            pack->msgid |= buff[reqoffset + 5] << 16;
            pack->msgid |= buff[reqoffset + 6] << 8;
            pack->msgid |= buff[reqoffset + 7] << 0;

            pack->sta = buff[reqoffset + 8] << 24;
            pack->sta |= buff[reqoffset + 9] << 16;
            pack->sta |= buff[reqoffset + 10] << 8;
            pack->sta |= buff[reqoffset + 11] << 0;

            if (pack->datalen) {
                pack->datapack = buff + offset + fixpackbase - 1;
            } else {
                pack->datapack = NULL;
            }

            return 0;
        }
    }
    return -2;
}

/*===================RPC API================*/

/**
 * @brief Init the artosyn rpc module.
 *
 * @param   handle  Refers to oal_handle.
 *
 * @return  N/A
 */
void *ar_rpc_init(void *handle)
{
    int ret = 0;
    void *rd_lock, *wr_lock;
    struct ar_rpc *rpc = NULL;
    oal_handle* phandle = (oal_handle*)handle;

    if (!handle) {
        return phandle;
    }

    AR_DBG(DM_RPC, DL_INF, AR_FN_ENTRY_STR);

    rpc = kzalloc(sizeof(struct ar_rpc), GFP_ATOMIC);
    if (rpc) {
        rpc->ar_handle = (void*)phandle;

        spin_lock_init(&phandle->ar_rpc_lock);
        atomic_set(&phandle->ar_rpc_ref_cnt, 0);
        spin_lock_init(&rpc->op_spinlock);

        rpc->rb = (ringbuffer_t *)kzalloc(sizeof(ringbuffer_t), GFP_ATOMIC);
        if (!rpc->rb) {
            kfree(rpc);
            return NULL;
        }

        rpc->rb->buffer = (u8*)vmalloc(RX_RINGBUFFER_SIZE);
        if (!rpc->rb->buffer) {
            kfree(rpc->rb);
            kfree(rpc);
            return NULL;
        }

        rpc->rb->size = RX_RINGBUFFER_SIZE;

        ringbuffer_reset(rpc->rb);

        /* Init wr spin lock */
        wr_lock = (void*)rpc->op_wr_spinlock;
        ret = oal_init_lock(phandle, &wr_lock);
        if (ret != 0) {
            AR_DBG(DM_RPC, DL_INF, "%s init wr spin lock failed! ret %d\n", __FUNCTION__, ret);
            goto fail;
        }

        /* Init rd spin lock */
        rd_lock = (void*)rpc->op_rd_spinlock;
        ret = oal_init_lock(phandle, &rd_lock);
        if (ret != 0) {
            AR_DBG(DM_RPC, DL_INF, "%s init rd spin lock failed! ret %d\n", __FUNCTION__, ret);
        }

        /** Init the socket list */
        INIT_LIST_HEAD(&rpc->session_socket_list);

        rpc->state = AR_RPC_STATE_UP;
    }

    AR_DBG(DM_RPC, DL_INF, AR_FN_LEAVE_STR);

    return (void*)rpc;

fail:
    if (rpc->op_rd_spinlock) {
        oal_free_lock(phandle, rpc->op_rd_spinlock);
    }

    if (rpc->op_wr_spinlock) {
        oal_free_lock(phandle, rpc->op_wr_spinlock);
    }

    if (rpc) {
        if (rpc->rb) {
            if (rpc->rb->buffer) {
                vfree(rpc->rb->buffer);
            }
            kfree(rpc->rb);
        }
        kfree(rpc);
    }

    AR_DBG(DM_RPC, DL_INF, AR_FN_LEAVE_STR);

    return NULL;
}

/**
 * @brief Deinit the artosyn rpc module.
 *
 * @param   handle  Refers to oal_handle.
 *
 * @return  N/A
 */
void ar_rpc_deinit(void *handle)
{
    oal_handle* phandle = (oal_handle*)handle;
    struct ar_rpc *rpc = NULL;
    unsigned long flags;

    if (!phandle) {
        return;
    }

    // free rpc resource
    spin_lock_irqsave(&phandle->ar_rpc_lock, flags);
    rpc = (struct ar_rpc*)phandle->ar_rpc;
    if (!rpc) {
        spin_unlock_irqrestore(&phandle->ar_rpc_lock, flags);
        return;
    }
    rpc->state = AR_RPC_STATE_DOWN;
    spin_unlock_irqrestore(&phandle->ar_rpc_lock, flags);

    // check the atomic cnt
    while (atomic_read(&phandle->ar_rpc_ref_cnt) != 0) {
        cpu_relax();  // relax cpu for less cost
    }

    if (rpc->rb) {
        if (rpc->rb->buffer) {
            vfree(rpc->rb->buffer);
        }
        kfree(rpc->rb);
    }
    kfree(rpc);
}

/**
 * @brief Set the state of artosyn rpc as stop.
 *
 * @param   handle  Refers to struct oal_handle.
 *
 * @return  N/A
 */
void ar_rpc_state_stop(void *handle)
{
    unsigned long flags;
    struct ar_rpc *rpc = NULL;
    oal_handle* phandle = (oal_handle*)handle;

    AR_DBG(DM_RPC, DL_INF, AR_FN_ENTRY_STR);

    if (phandle) {
        return;
    }

    spin_lock_irqsave(&phandle->ar_rpc_lock, flags);
    rpc = (struct ar_rpc *)phandle->ar_rpc;
    if (rpc == NULL) {
        AR_DBG(DM_RPC, DL_ERR, "%s: rpc is null!\n", __func__);
        spin_unlock_irqrestore(&phandle->ar_rpc_lock, flags);
        return;
    }

    // stop ipc to prevent usb data to/from hawk_usb module
    AR_DBG(DM_RPC, DL_INF, "%s %d\n", __func__, __LINE__);
    rpc->state = AR_RPC_STATE_STOP;
    spin_unlock_irqrestore(&phandle->ar_rpc_lock, flags);
}

/**
 * @brief Set the state of artosyn rpc as up.
 *
 * @param   handle  Refers to struct oal_handle.
 *
 * @return  N/A
 */
void ar_rpc_state_up(void *handle)
{
    unsigned long flags;
    struct ar_rpc *rpc = NULL;
    oal_handle* phandle = (oal_handle*)handle;

    AR_DBG(DM_RPC, DL_INF, AR_FN_ENTRY_STR);

    if (phandle) {
        return;
    }

    spin_lock_irqsave(&phandle->ar_rpc_lock, flags);

    rpc = (struct ar_rpc *)phandle->ar_rpc;
    if (rpc == NULL) {
        AR_DBG(DM_RPC, DL_ERR, "%s: rpc is null!\n", __func__);
        spin_unlock_irqrestore(&phandle->ar_rpc_lock, flags);
        return;
    }

    // stop ipc to prevent usb data to/from hawk_usb module
    AR_DBG(DM_RPC, DL_INF, "%s %d\n", __func__, __LINE__);

    rpc->state = AR_RPC_STATE_UP;
    spin_unlock_irqrestore(&phandle->ar_rpc_lock, flags);
}

/**
 * @brief Analyze the pkt and dump the info.
 *
 * @param   ar_rpc  Refers to struct ar_rpc.
 * @param   skb     skb with data
 *
 * @return  N/A
 */
void ar_rpc_hdr_print(void *ar_rpc, struct sk_buff *skb)
{
    int ret;
    rpc_pack_t pack;

    memset(&pack, 0, sizeof(rpc_pack_t));

    /*analyze ipc hdr*/
    ret = unpack_usb_pack(skb->data, skb->len, NULL, &pack);
    if (ret < 0) {
        return;
    }

    dump_bytes(DL_INF, "rpc hdr", skb->data, data_offset);
}

/**
 * @brief Unpack the pkt and dispatch the skb to net_dev or other module.
 *
 * @param   ar_rpc  Refers to struct ar_rpc.
 * @param   skb     skb with data
 *
 * @return  0 for success, otherwise fail. If ret < 0, the caller should
 *          free the skb, otherwise means skb has been freed.
 */
static int
ar_rpc_rx_pkt_schedule_task(struct ar_rpc *rpc, struct sk_buff *skb)
{
    int ret = 0;
    int len = 0;
    int en_buf_len = 0;
    int usedlen = 0;
    int start_offset = 0;
    int buff_len = 0;
    int sock_slot = 0;
    int sock_port = 0;
    u8 *data = NULL;
    rpc_pack_t pack;
    struct sk_buff *skb_rxpkt = NULL;
    u8 so_cmd = 0;
    u8 req_type = 0;
    oal_handle *handle = (oal_handle*)rpc->ar_handle;
    ar_netif_t netif = {0}; // for intf lookup
    unsigned long flags;

    AR_DBG(DM_RPC, DL_INF, AR_FN_ENTRY_STR);

    if (!handle) {
        AR_DBG(DM_RPC, DL_ERR, "%s: handle is NULL!\n", __func__);
        AR_DBG(DM_RPC, DL_INF, AR_FN_LEAVE_STR);
        return -ENODEV;
    }

    if (!skb) {
        AR_DBG(DM_RPC, DL_ERR, "%s: skb is NULL!\n", __func__);
        AR_DBG(DM_RPC, DL_INF, AR_FN_LEAVE_STR);
        return -ESRCH;
    }

    oal_spin_lock(handle, rpc->op_rd_spinlock);

    data = skb->data;
    len = skb->len;
    /*analyze ipc hdr*/
    memset(&pack, 0, sizeof(rpc_pack_t));
    ret = unpack_usb_pack(data, len, &usedlen, &pack);
    if (ret < 0) {
        AR_DBG(DM_RPC, DL_ERR, "%s: Unpack data failed!\n", __func__);
        AR_DBG(DM_RPC, DL_INF, AR_FN_LEAVE_STR);
        return ret;
    }

    so_cmd = (pack.reqid >> 24) & 0xff;
    req_type = (pack.reqid >> 16) & 0xff;
    sock_slot = (pack.reqid >> 8) & 0xff;
    sock_port = (pack.reqid >> 0) & 0xff;

    AR_DBG(DM_RPC, DL_INF, "%s so_cmd %d req_type %d slot %d port %d\n"
            , __FUNCTION__, so_cmd, req_type, sock_slot, sock_port);

    switch (so_cmd) {
        case 4://socket
        {
            /* Get the netif with slot+port */
            netif.type = AR_NET_NETIF_T_PORT;
            netif.socket_port = sock_port;
            netif.slot = sock_slot;

            if (!ar_net_exists(handle, &netif)) {
                mdev_spin_lock((&handle->ar_drv_mdev), flags);
                /* Send skb to user if there's no netif to send */
                skb_queue_tail(&handle->ar_drv_mdev.rx_q, skb);
                wake_up_interruptible(&handle->ar_drv_mdev.req_wait_q);
                mdev_spin_unlock((&handle->ar_drv_mdev), flags);
                break;
            }

            switch(req_type) {
                case so_read:
                    /** remove the padding data */
                    skb_trim(skb, usedlen);
                    /*remove rpc hdr*/
                    skb_pull(skb, data_offset);
                    /*remove rpc tail*/
                    skb_trim(skb, skb->len - (fixpackbase - data_offset));

                    if (ar_net_is_datagram_socket(handle, &netif)) { //with datagram
                        AR_DBG(DM_RPC, DL_VRB, "%s %d push into ringbuffer\n",__FUNCTION__,__LINE__);
                        len = skb->len;

                        dump_bytes(DL_VRB, "rx", skb->data, skb->len);

                        en_buf_len = ringbuffer_in(rpc->rb, skb->data, len);
                        /* Release the tmp skb */
                        dev_kfree_skb_any(skb);
                        if (en_buf_len != len) {
                            AR_DBG(DM_RPC, DL_WRN, "%s %d out of mem!! len %d\n",__FUNCTION__,__LINE__, len);
                            ringbuffer_out(rpc->rb, NULL, en_buf_len);
                            break;
                        }

                        do {
                            ret = unpack_socket_datagram_pack_rb(rpc->rb, &start_offset, &buff_len);
                            if (ret != 0) {
                                /*free ringbuffer*/
                                AR_DBG(DM_RPC, DL_VRB, "%s %d datagram is not complete\n",__FUNCTION__,__LINE__);
                                /** skb has been processd */
                                ret = 0;
                                break;
                            }
                            else {
                                AR_DBG(DM_RPC, DL_VRB, "%s %d datagram is ok start_offset %d buff_len %d\n",
                                        __FUNCTION__,__LINE__, start_offset, buff_len);

                                /* Drop err head */
                                ringbuffer_out(rpc->rb, NULL, start_offset);

                                /* Alloc the rx buffer */
                                skb_rxpkt = dev_alloc_skb(buff_len);
                                if (!skb_rxpkt) {
                                    AR_DBG(DM_RPC, DL_WRN, "sokcet datagram pkt alloc skb failed!\n");
                                    break;
                                }

                                /*copy data*/
                                skb_put(skb_rxpkt, buff_len);
                                ringbuffer_out(rpc->rb, skb_rxpkt->data, buff_len);

                                /* remove datagram*/
                                /* remove datagram hdr*/
                                skb_pull(skb_rxpkt, datagram_offset);
                                /* remove datagram tail*/
                                skb_trim(skb_rxpkt, skb_rxpkt->len - (datagram_len - datagram_offset));

                                dump_bytes(DL_VRB, "rx to net", skb_rxpkt->data, skb_rxpkt->len);
                            }

                            ret = ar_net_rx_enqueue(handle, &netif, skb_rxpkt);
                            if (ret < 0) {
                                if (skb_rxpkt) {
                                    dev_kfree_skb_any(skb_rxpkt);
                                }
                                /* skb has been freed */
                                ret = 0;
                                AR_DBG(DM_RPC, DL_ERR, "%s: ar_net_rx_enqueue failed. ret %d\n", __func__, ret);
                            }

                            skb_rxpkt = NULL;
                        } while(true);
                    }
                    else {
                        ret = ar_net_rx_enqueue(handle, &netif, skb);
                        if (ret < 0) {
                            AR_DBG(DM_RPC, DL_ERR, "%s: ar_net_rx_enqueue failed. ret %d\n", __func__, ret);
                        }
                    }

                    break;
                case so_write:
                    {
                        struct socket_msg_ret retmsg = {0};
                        AR_DBG(DM_RPC, DL_VRB, "%s: so_write ack\n", __func__);
                        if (pack.datapack) {
                            memcpy(&retmsg, pack.datapack, sizeof(retmsg));
                        }

                        /** Inform the socket session */
                        socket_tx_ack_inform(handle, sock_slot, sock_port, pack.sta, &retmsg);

                        /* Release the skb */
                        dev_kfree_skb_any(skb);
                    }
                    break;
                default:
                    AR_DBG(DM_RPC, DL_VRB, "%s: unknown ack\n", __func__);
                    /* Release the skb */
                    dev_kfree_skb_any(skb);
                    break;
            }
            break;
        }
        default:
            dump_bytes(DL_VRB, "PKT transparent", skb->data, skb->len);
            mdev_spin_lock((&handle->ar_drv_mdev), flags);
            skb_queue_tail(&handle->ar_drv_mdev.rx_q, skb);
            wake_up_interruptible(&handle->ar_drv_mdev.req_wait_q);
            mdev_spin_unlock((&handle->ar_drv_mdev), flags);
            break;
    }

    oal_spin_unlock(handle, rpc->op_rd_spinlock);

    AR_DBG(DM_RPC, DL_INF, AR_FN_LEAVE_STR);

    return ret;
}

/**
 * @brief Process the skb and put to rx queue of net dev.
 *
 * @param   handle  Refers to struct oal_handle.
 * @param   skb     skb with data
 *
 * @return  N/A
 */
void ar_rpc_rx_pkt(void *handle, struct sk_buff *skb)
{
    oal_handle* phandle = (oal_handle*)handle;
    struct ar_rpc *rpc = NULL;
    unsigned long flags;
    u32 buf_len = 0;
    int ret = 0;

    /* 1a. check input */
    if (!skb) {
        AR_DBG(DM_RPC, DL_ERR, "%s: skb is null\n", __func__);
        goto err_oops;
    }

    dump_bytes(DL_VRB, "rx pkt", skb->data, skb->len);

    spin_lock_irqsave(&phandle->ar_rpc_lock, flags);
    rpc = (struct ar_rpc *)phandle->ar_rpc;
    if (!rpc) {
        AR_DBG(DM_RPC, DL_ERR, "%s: rpc is null!\n", __func__);
        WARN_ON(skb->next);
        dev_kfree_skb_any(skb);
        spin_unlock_irqrestore(&phandle->ar_rpc_lock, flags);
        return;
    }

    /* 1b. check state */
    if ((rpc->state == AR_RPC_STATE_STOP) || (rpc->state == AR_RPC_STATE_DOWN)) {
        AR_DBG(DM_RPC, DL_ERR, "%s: rpc is stoped!\n", __func__);
        WARN_ON(skb->next);
        dev_kfree_skb_any(skb);
        spin_unlock_irqrestore(&phandle->ar_rpc_lock, flags);
        return;
    }
    atomic_inc(&phandle->ar_rpc_ref_cnt);
    spin_unlock_irqrestore(&phandle->ar_rpc_lock, flags);

    /* 2. put to rx queue */
    ret = ar_rpc_rx_pkt_schedule_task(rpc, skb);
    if (ret != 0) {
        AR_DBG(DM_RPC, DL_WRN, "%s: ar_rpc_rx_pkt_schedule_task failed! ret %d\n", __func__, ret);
        dev_kfree_skb_any(skb);
        atomic_dec(&phandle->ar_rpc_ref_cnt);
        return;
    }

    atomic_dec(&phandle->ar_rpc_ref_cnt);

    return;

err_oops:
    BUG();
}

/**
 * @brief Assemble the rpc pkt and send to bus.
 *
 * @param   phandle     Refers to oal_handle.
 * @param   pack        rpc_pack_t with rpc info
 * @param   skb         skb with data
 * @param   priv_data   For free the data.
 *
 * @return  0 for success, otherwise fail.
 */
int ar_rpc_tx_pkt(void* phandle, rpc_pack_t *pack, struct sk_buff *skb, void *priv_data)
{
    int ret = 0;
    u8 *data;
    int len = 0;
    int sdlen = 0;
    oal_status status = OAL_STATUS_SUCCESS;
    oal_handle* handle = (oal_handle*)phandle;

    if (!handle) {
        AR_DBG(DM_RPC, DL_ERR, "%s:handle is NULL!\n", __FUNCTION__);
        return -EPERM;
    }

    if (!skb) {
        AR_DBG(DM_RPC, DL_ERR, "skb is NULL!\n");
        return -ENOMEM;
    }

    len = skb->len;

    /*add rpc hdr*/
    skb_push(skb, data_offset);
    /*add rpc tail*/
    skb_put(skb, fixpackbase - data_offset);
    data = (u8 *)skb->data;
    /* fill the header */
    make_rpc_pack2buff_no_copy(data, len, pack);
    sdlen = skb->len;

    dump_bytes(DL_VRB, "tx pkt", skb->data, skb->len);

    status = oal_send_packet_req(handle, skb, priv_data);
    if (status != OAL_STATUS_SUCCESS) {
        ret = -EIO;
        AR_DBG(DM_RPC, DL_VRB, "%s: oal_send_packet_req failed, reqid=0x%x, status=%d\n", __func__, pack->reqid, status);
        return ret;
    }

    return sdlen;
}

/**
 * @brief Assemble the rpc msg and send to bus.
 *
 * @param   phandle Refers to oal_handle.
 * @param   pack    rpc_pack_t with rpc info
 * @param   skb     skb with data
 *
 * @return  0 for success, otherwise fail.
 */
int ar_rpc_tx_msg(void* phandle, rpc_pack_t *pack, struct sk_buff *skb)
{
    int ret = 0;
    u8 *data = NULL;
    int sdlen = 0;
    int len = 0;
    oal_status status = OAL_STATUS_SUCCESS;
    oal_handle* handle = (oal_handle*)phandle;

    AR_DBG(DM_RPC, DL_VRB, AR_FN_ENTRY_STR);

    if (!handle) {
        AR_DBG(DM_RPC, DL_ERR, "%s:handle is NULL!\n", __FUNCTION__);
        return -1;
    }

    if (!skb) {
        return -ENOMEM;
    }

    // if (pack->datalen != 0 && pack->datapack != NULL) {
    //     data = (u8 *) skb_put(skb, pack->datalen);
    //     memcpy(data, pack->datapack, pack->datalen);
    // }

    len = skb->len;

    /*add hdr*/
    skb_push(skb, data_offset);
    /*add tail*/
    skb_put(skb, fixpackbase - data_offset);

    data = (u8 *)skb->data;

    /* fill the header */
    make_rpc_pack2buff_no_copy(data, len, pack);

    dump_bytes(DL_VRB, "tx msg", skb->data, skb->len);

    ret = oal_send_msg_req(handle, skb, false);
    if (status != OAL_STATUS_SUCCESS) {
        ret = -EIO;
        AR_DBG(DM_RPC, DL_ERR, "%s: tx failed, reqid=0x%x, ret=%d\n", __func__, pack->reqid, ret);
    }

    AR_DBG(DM_RPC, DL_VRB, AR_FN_LEAVE_STR);

    return skb->len;
}
